package com.atguigu.chapter05.sink;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList;

/**
 * Author: Pepsi
 * Date: 2023/7/29
 * Desc:
 */
public class Flink06_Sink_Custom_Mysql {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 1000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        ArrayList<WaterSensor> waterSensors = new ArrayList<>();
        waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
        waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
        waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
        waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
        waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));


        DataStreamSource<WaterSensor> stream = env.fromCollection(waterSensors);

        SingleOutputStreamOperator<WaterSensor> result = stream
                .keyBy(WaterSensor::getId)
                .sum("vc");

        result.addSink(new MySqlSink());


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public static class MySqlSink extends RichSinkFunction<WaterSensor> {

        private Connection connection;

        @Override
        public void open(Configuration parameters) throws Exception {
            Class.forName("com.mysql.cj.jdbc.Driver");
            connection = DriverManager.getConnection("jdbc:mysql://hadoop101:3306/test?useSSL=false", "root", "000000");
        }

        @Override
        public void close() throws Exception {
            if (connection!=null){
                connection.close();
            }
        }

        // 调用：每来一条元素，这个方法执行一次
        @Override
        public void invoke(WaterSensor value, Context context) throws Exception {
            // jdbc的方式想MySQL写数据
//            String sql = "insert into sensor(id,ts,vc)values(?,?,?)";
            //如果主键不重复就新增，主键重复就更新
//            String sql = "insert into sensor(id,ts,vc)values(?,?,?) duplicate key update vc=?";
            String sql = "replace into sensor(id,ts,vc)values(?,?,?)";
            // 1. 得到预处理语句
            PreparedStatement ps = connection.prepareStatement(sql);
            // 2. 给sql中的占位符进行赋值
            ps.setString(1,value.getId());
            ps.setLong(2,value.getTs());
            ps.setInt(3,value.getVc());
//            ps.setInt(4,value.getVc());
            // 3. 执行
            ps.execute();
            // 4. 提交
//            connection.commit();  MySQL默认自动提交，所以这个地方不用调用
            // 5. 关闭预处理
            ps.close();
        }
    }
}
